package com.qlslylq.ad.sdk.rx;

import com.qlslylq.ad.sdk.rx.subscribers.OnNext;
import com.qlslylq.ad.sdk.rx.subscribers.QuietSubscriber;
import com.qlslylq.ad.sdk.rx.subscribers.Subscriber;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.schedulers.Schedulers;

/**
 * Rx工具类<br/>
 */
public class RxUtils {


    /**
     * 后台默认加载数据,返回成功
     *
     * @param flowable
     * @param nextListener
     * @param <T>
     */
    public static <T> Disposable rx(Flowable<T> flowable, OnNext<T> nextListener) {
        return flowable
                .onBackpressureDrop()  //加上背压策略
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new QuietSubscriber<>(nextListener));
    }

    /**
     * 后台默认加载数据,返回成功
     *
     * @param flowable
     * @param nextListener
     * @param <T>
     */
    public static <T> Disposable rxThird(Flowable<T> flowable, OnNext<T> nextListener) {
        return flowable
                .onBackpressureDrop()  //加上背压策略
                .subscribeOn(Schedulers.io())
                .unsubscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribeWith(new Subscriber<>(nextListener));
    }

    /**
     * 异步任务
     */
    public static <T> Disposable rx(ObservableOnSubscribe<T> source, Consumer<? super T> onNext) {
        return Observable.create(source)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(onNext);
    }

}